package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.pojo.vo.ChargingProcessVO;
import cn.tedu.charging.common.uitls.JsonUitls;
import cn.tedu.charging.common.web.JsonResult;
import cn.tedu.charging.order.constant.Constant;
import cn.tedu.charging.order.constant.MqttConstant;
import cn.tedu.charging.order.constant.OrderStatusConstant;
import cn.tedu.charging.order.dao.repository.ChargingProcessRepository;
import cn.tedu.charging.order.dao.repository.OrderRepository;
import cn.tedu.charging.common.pojo.param.ChargingProcessParam;
import cn.tedu.charging.order.feign.CostClient;
import cn.tedu.charging.order.feign.UserClient;
import cn.tedu.charging.order.pojo.param.ChargingResultParam;
import cn.tedu.charging.order.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.order.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.order.pojo.po.ChargingProcessPO;
import cn.tedu.charging.order.pojo.vo.CharingProcessVO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.math.BigDecimal;

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {


    private MqttClient mqttClient;
    private MqttContext mqttContext;

    public MqttConsumer(MqttClient mqttClient,MqttContext mqttContext) {
        this.mqttClient=mqttClient;
        this.mqttContext=mqttContext;

    }

    @Override
    public void connectComplete(boolean b, String s) {
        log.debug("MqttConsumer:连接成功");
        try {
            //订阅服务,服务名(/topic/start/result)
            //这个是服务的接收者
            //接收数据: 判断充电订单是否成功
            //和EMQ连接成功后,订阅/topic/start/result 来消费充电桩发送 充电是否成功 的指令
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_RESULT,0);
            //EMQ连接成功后,订阅/topic/charging/process 来消费充电桩发送 充电数据 的指令
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_PROCESS,0);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void connectionLost(Throwable throwable) {
        log.debug("MqttConsumer:连接失败");
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("MqttConsumer:消息到达:topic-{},message-{}", topic, message);
        if (MqttConstant.TOPIC_CHARGING_RESULT.equals(topic)){
            handleChargingResult(message);
        }else if(MqttConstant.TOPIC_CHARGING_PROCESS.equals(topic)){
            handleChargingProcess(message);
        }

    }

    //处理设备同步充电数据
    private void handleChargingProcess(MqttMessage message) {
        log.debug("MqttConsumer:消息到达 handleChargingProcess:message-{}",message);
        //把JSON转化为对象
        ChargingProcessParam chargingProcessParam = JsonUitls.fromJson(message.toString(), ChargingProcessParam.class);

        //1. 计算充电金额
        BigDecimal orderCost = getOrderCost(chargingProcessParam);
        //2.判断温度 大于45度 给用户推送(websocket)B级告警 大于50 推送 A级告警
        Float temperature = chargingProcessParam.getTemperature();
        if (temperature > MqttConstant.A_WARNING){ //50定义为常量
            log.debug("温度超过:{} 给用户发送A级告警,实际温度:{}",MqttConstant.A_WARNING,temperature);
            sendMessageByWebSocket("温度超过" + MqttConstant.A_WARNING + "给用户发送A级告警通知,直接停止设备,实际温度" + temperature,chargingProcessParam.getUserId());
        }else if (temperature > MqttConstant.B_WARNING){
            log.debug("温度超过:{} 给用户发送B级告警,实际温度:{}",MqttConstant.A_WARNING,temperature);
            sendMessageByWebSocket("温度超过" + MqttConstant.B_WARNING + "给用户发送B级告警通知,直接停止设备,实际温度" + temperature,chargingProcessParam.getUserId());
        }

        //3.判断电费是否大于金额
        //获取当前用户余额
        BigDecimal userBalance = getUserBalance(chargingProcessParam.getUserId());
        if (userBalance.subtract(orderCost).floatValue() <= 0) {
            log.debug("给用户发通知 余额不足,当前余额:{},订单消费:{}",userBalance,orderCost);
        }
        //4.判断是否充满 TODO
        //5. 推送(websocket)充电进度到手机端
        syncChargingProcess2User(chargingProcessParam.getUserId());

        //6.充电进度保存
        saveChargingProcess(chargingProcessParam);

    }

    //通过WebSokcet给用户发送消息
    private void sendMessageByWebSocket(String message,Integer userId){
        try {
            mqttContext.getWebSocketServer().sendInfo(message,String.valueOf(userId));
            log.debug("通过websocket发送消息成功：{}",message);
        } catch (IOException e) {
            log.error("通过websocket发送消息失败",e);
        }
    }

    //保存充电进度数据 设备同步数据(同步到:: ES 数据库 )
    private void saveChargingProcess(ChargingProcessParam chargingProcessParam) {
        log.debug("保存充电进度数据:{}",chargingProcessParam);
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        //惊醒时序排序,
        chargingProcessPO.setId(System.currentTimeMillis());
        BeanUtils.copyProperties(chargingProcessParam,chargingProcessPO);
        mqttContext.getChargingProcessRepository().save(chargingProcessPO);
        log.debug("保存充电进度数据成功:{}",chargingProcessPO);

    }

    //给用户同步充电数据
    private void syncChargingProcess2User(Integer userId) {
        log.debug("推送充电进入到手机端入参 userId ={}",userId);
        CharingProcessVO charingProcessVO = getChargingProcess();
        log.debug("给用户同步充电进度:{}",charingProcessVO);

    }

    //获取充电进度
    private CharingProcessVO getChargingProcess() {
        CharingProcessVO charingProcessVO = new CharingProcessVO();
        charingProcessVO.setChargingCapacity(1f);
        charingProcessVO.setTotalCapacity(95f);
        return charingProcessVO;
    }

    //获取用户余额
    private BigDecimal getUserBalance(Integer userId) {
        log.debug("获取用户余额信息userId={}",userId);
        JsonResult<BigDecimal> userBalance = mqttContext.getUserClient().getUserBalance(userId);
        log.debug("调用用户服务获取用户余额:入参:{},出参:{}",userId,userBalance);
        if (userBalance != null) {
            return userBalance.getData();
        }
        return new BigDecimal(0);
    }

    //获取电费
    private BigDecimal getOrderCost(ChargingProcessParam chargingProcessParam) {
        log.debug("计算电费入参 orderNo:{}",chargingProcessParam);
        JsonResult<ChargingProcessVO> result = mqttContext.getCostClient().calculateCost(chargingProcessParam);
        if (result != null) {
            ChargingProcessVO data = result.getData();
            log.debug("计算电费入参 orderNo:{} 出参:{}",chargingProcessParam,data);
            if (data !=null){
                return data.getTotalCost();
            }
        }
        return new BigDecimal(0);
    }

    //处理设备同步的充电结果
    private void handleChargingResult(MqttMessage message) {
        log.debug("MqttConsumer:消息到达 handleChargingResult:message-{}",message);
        ChargingResultParam resultParam = JsonUitls.fromJson(message.toString(), ChargingResultParam.class);
        log.debug("ChargingResultParam转化后的数据 : resultParam:{}",resultParam);
        if (resultParam!=null){
            String result = resultParam.getResult();
            if (Constant.RESULT_START_CHARGING_SUCCESS.equals(result)){
                log.debug("充电桩开始充电");
                //创建订单成功记录
                ChargingBillSuccessPO chargingBillSuccessPO = new ChargingBillSuccessPO();
                chargingBillSuccessPO.setBillId(resultParam.getOrderNo());
                chargingBillSuccessPO.setUserId(resultParam.getUserId());
                //订单状态变更为充电中
                chargingBillSuccessPO.setBillStatus(OrderStatusConstant.ORDER_STATUS_PROGRESS);
                log.debug("INSERT订单的数据-{}",chargingBillSuccessPO);
                Integer row = mqttContext.getOrderRepository().saveSuccessOrder(chargingBillSuccessPO);
                log.debug("INSERT订单数数据成功:{},影响行数:{}",chargingBillSuccessPO,row);
            }else if (Constant.RESULT_START_CHARGING_FAIL.equals(result)) {
                log.debug("充电桩故障");
                //创建订单失败记录
                ChargingBillFailPO chargingBillFailPO = new ChargingBillFailPO();
                chargingBillFailPO.setFailDesc("设备自检失败-设备有响应但是充不了电");
                chargingBillFailPO.setBillId(resultParam.getOrderNo());
                mqttContext.getOrderRepository().saveFailOrder(chargingBillFailPO);
            }
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.debug("MqttConsumer:消息处理完成");
    }
}
